package net.bwie.realtime.jtp.ads.trade.job;

import net.bwie.realtime.jtp.ads.trade.bean.Order;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;
import java.util.Random;

/**
 * 实时统计每个主播的成交总额
 */
public class DouyinLiveGMV {
    public static void main(String[] args) throws Exception {
        // 1. 初始化执行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4); // 设置并行度（机器越多可调越大）

        // 2. 模拟订单数据源
        DataStream<Order> orders = env.fromElements(
                //orderId，userId，anchorId，itemId，price，ts
                new Order("o1", "u1", "anchorA", "item01", 100.0, 1000L),
                new Order("o2", "u2", "anchorB", "item02", 200.0, 2000L),
                new Order("o3", "u3", "anchorA", "item03", 300.0, 3000L),
                new Order("o4", "u4", "anchorHot", "item04", 400.0, 4000L),
                new Order("o5", "u5", "anchorHot", "item05", 500.0, 5000L)
        ).assignTimestampsAndWatermarks(
                WatermarkStrategy.<Order>forBoundedOutOfOrderness(Duration.ofSeconds(5)) // 允许5秒乱序
                        .withTimestampAssigner((SerializableTimestampAssigner<Order>) (order, recordTs) -> order.ts)
        );

        // 3. 数据倾斜优化：给 anchor_id 加随机前缀
        DataStream<Tuple2<String, Double>> anWithPrefix = orders
                .map((MapFunction<Order, Tuple2<String, Double>>) order -> {
                    int prefix = new Random().nextInt(10); // 随机0-9，10个数
                    return new Tuple2<>(prefix + "_" + order.anchorId, order.price);
                });

        // 4. 局部聚合（避免单个主播数据过载）
        DataStream<Tuple2<String, Double>> partialAgg = anWithPrefix
                .keyBy(value -> value.f0) // 按随机前缀+主播ID分组
                .window(TumblingEventTimeWindows.of(Time.minutes(1))) // 开1分钟窗口
                .sum(1); // 对GMV进行统计

        // 5. 二次聚合按主播ID全局汇总
        DataStream<Tuple2<String, Double>> finalAgg = partialAgg
                .map((MapFunction<Tuple2<String, Double>, Tuple2<String, Double>>) value -> {
                    String anchorId = value.f0.split("_")[1]; // 去掉随机前缀
                    return new Tuple2<>(anchorId, value.f1);
                })
                .keyBy(value -> value.f0) // 按主播ID汇总
                .window(TumblingEventTimeWindows.of(Time.minutes(1)))
                .sum(1);

        // 6. 输出结果
        finalAgg.print("finalAgg===============>");

        env.execute("DouyinLiveGMV");
    }
}

